package cn.sh.ideal.yedis.core;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import cn.sh.ideal.yedis.util.TimeUnit;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPool;
import redis.clients.util.Pool;
/**
 * 集群操作
 * @author WYY
 *
 * @param <V>
 */
@SuppressWarnings("unchecked")
public class ClusterOperation<V> extends AbstractOperation<V> implements RedisOperation<V> {

	public ClusterOperation(RedisTemplate<V> template) {
		super(template);
	}

	@Override
	public void set(final String key, final V value) {
		executeCluster(new RedisCallback<JedisCluster, Void>() {
			@Override
			public Void doInRedis(JedisCluster connection) {
				if(null == value) return null;
				connection.set(rawKey(key),raw(value));
				return null;
			}

		});
	}

	@Override
	public void set(final String key, final V value, final int timeout, final TimeUnit unit) {
		executeCluster(new RedisCallback<JedisCluster, Void>() {
			@Override
			public Void doInRedis(JedisCluster connection) {
				if(null == value) return null;
				connection.setex(rawKey(key), timeout * unit.getCount(), raw(value));
				return null;
			}
		});

	}

	@Override
	public boolean setNx(final String key, final V value) {
		return executeCluster(new RedisCallback<JedisCluster, Boolean>() {
			@Override
			public Boolean doInRedis(JedisCluster connection) {
				if(null == value) return false;
				return connection.setnx(rawKey(key),raw(value)) == 1;
			}
		});
	}

	@Override
	public boolean set(final String key, final String value, final String nxxx,final long timeout, final String expx) {
		return executeCluster(new RedisCallback<JedisCluster, Boolean>() {
			@Override
			public Boolean doInRedis(JedisCluster connection) {
				if(null == value) return false;
				String result = connection.set(key,value,nxxx,expx,timeout);
				return STATUS_SUCCESS.equals(result);
			}
		});
	}

	@Override
	public V get(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, V>() {
			@Override
			public V doInRedis(JedisCluster connection) {
				return convert(connection.get(rawKey(key)));
			}
		});
	}

	@Override
	public Long lpush(final String key, final V... values) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				if(null == values || values.length == 0) return 0l; 
				return connection.lpush(rawKey(key), toByteArray(values));
			}
		});
	}

	@Override
	public Long rpush(final String key, final V... values) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				if(null == values || values.length == 0) return 0l; 

				return connection.rpush(rawKey(key), toByteArray(values));
			}
		});
	}

	@Override
	public V lpop(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, V>() {
			@Override
			public V doInRedis(JedisCluster connection) {
				return convert(connection.lpop(rawKey(key)));
			}
		});
	}

	@Override
	public V rpop(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, V>() {
			@Override
			public V doInRedis(JedisCluster connection) {
				return convert(connection.rpop(rawKey(key)));
			}
		});
	}

	@Override
	public V blpop(final int timeout,final String key) {
		return executeCluster(new RedisCallback<JedisCluster, V>() {
			@Override
			public V doInRedis(JedisCluster connection) {
				List<byte[]> list = connection.blpop(timeout, rawKey(key));
				if(null == list) {
					return null;
				}else {
					return convert(list.get(1));
				}
			}
		});
	}

	@Override
	public V brpop(final int timeout,final String key) {
		return executeCluster(new RedisCallback<JedisCluster, V>() {
			@Override
			public V doInRedis(JedisCluster connection) {
				List<byte[]> list = connection.brpop(timeout, rawKey(key));
				if(null == list) {
					return null;
				}else {
					return convert(list.get(1));
				}
			}
		});
	}

	@Override
	public Long llen(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.llen(key);
			}
		});
	}

	@Override
	public List<V> lrange(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, List<V>>() {
			@Override
			public List<V> doInRedis(JedisCluster connection) {
				return lrange(key, 0l, llen(key));
			}
		});
	}

	@Override
	public List<V> lrange(final String key,final Long start,final Long end) {
		return executeCluster(new RedisCallback<JedisCluster, List<V>>() {
			@Override
			public List<V> doInRedis(JedisCluster connection) {
				List<byte[]> list = connection.lrange(rawKey(key), start, end);
				return toObjectList(list);
			}
		});

	}

	@Override
	public V lindex(final String key,final Long index) {
		return executeCluster(new RedisCallback<JedisCluster, V>() {
			@Override
			public V doInRedis(JedisCluster connection) {
				return convert(connection.lindex(rawKey(key), index));
			}
		});
	}

	@Override
	public Long lrem(final String key, final V value) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.lrem(rawKey(key), 0, raw(value));
			}
		});
	}

	@Override
	public Long hset(final String key, final String field, final V value) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				if(null == field || null == value) return 0l;
				return connection.hset(rawKey(key), rawKey(field), raw(value));
			}
		});
	}

	@Override
	public String ltrim(final String key, final long start, final long end) {
		return executeCluster(new RedisCallback<JedisCluster, String>() {
			@Override
			public String doInRedis(JedisCluster connection) {
				return connection.ltrim(key, start, end);
			}
		});
	}

	@Override
	public Long hsetnx(final String key, final String field, final V value) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				if(null == field || null == value) return 0l;
				return connection.hsetnx(rawKey(key), rawKey(field), raw(value));
			}
		});
	}

	@Override
	public V hget(final String key, final String field) {
		return executeCluster(new RedisCallback<JedisCluster, V>() {
			@Override
			public V doInRedis(JedisCluster connection) {
				if(null == field) return null;
				return convert(connection.hget(rawKey(key), rawKey(field)));
			}
		});
	}

	@Override
	public String hmset(final String key, final Map<String, V> hash) {
		return executeCluster(new RedisCallback<JedisCluster, String>() {
			@Override
			public String doInRedis(JedisCluster connection) {
				if(hash == null || hash.isEmpty()) return null;

				return connection.hmset(rawKey(key), toByteMap(hash));
			}
		});
	}

	@Override
	public List<V> hmget(final String key, final String... fields) {
		return executeCluster(new RedisCallback<JedisCluster, List<V>>() {
			@Override
			public List<V> doInRedis(JedisCluster connection) {
				List<V> destList = new ArrayList<>();
				if(fields != null && fields.length > 0) {
					List<byte[]> slist = connection.hmget(rawKey(key), toByteArray(fields));
					slist.removeAll(Collections.singleton(null));
					destList = toObjectList(slist);
				}
				return destList;
			}
		});
	}

	@Override
	public Boolean hexists(final String key, final String field) {
		return executeCluster(new RedisCallback<JedisCluster, Boolean>() {
			@Override
			public Boolean doInRedis(JedisCluster connection) {
				if(null == field) return false;
				return connection.hexists(key, field);
			}
		});
	}

	@Override
	public Long hdel(final String key, final String... field) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				if(null == field || field.length == 0) return 0l;
				return connection.hdel(key, field);
			}
		});
	}

	@Override
	public Long hlen(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.hlen(key);
			}
		});
	}

	@Override
	public Set<String> hkeys(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, Set<String>>() {
			@Override
			public Set<String> doInRedis(JedisCluster connection) {
				return connection.hkeys(key);
			}
		});
	}

	@Override
	public List<V> hvals(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, List<V>>() {
			@Override
			public List<V> doInRedis(JedisCluster connection) {
				return toObjectList(connection.hvals(rawKey(key)));
			}
		});
	}

	@Override
	public Map<String, V> hgetAll(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, Map<String,V>>() {
			@Override
			public Map<String,V> doInRedis(JedisCluster connection) {
				return toObjectMap(connection.hgetAll(rawKey(key)));
			}
		});
	}

	@Override
	public Long sadd(final String key, final V... member) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				if(null == member || member.length == 0) return 0l;

				return connection.sadd(rawKey(key), toByteArray(member));
			}
		});
	}

	@Override
	public Set<V> smembers(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, Set<V>>() {
			@Override
			public Set<V> doInRedis(JedisCluster connection) {
				return toObjectSet(connection.smembers(rawKey(key)));
			}
		});
	}

	@Override
	public Long srem(final String key, final V... member) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				if(null == member || member.length == 0) return 0l;
				return connection.srem(rawKey(key), toByteArray(member));
			}
		});
	}

	@Override
	public V spop(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, V>() {
			@Override
			public V doInRedis(JedisCluster connection) {
				return convert(connection.spop(rawKey(key)));
			}
		});
	}

	@Override
	public Set<V> spop(final String key, final Long count) {
		return executeCluster(new RedisCallback<JedisCluster, Set<V>>() {
			@Override
			public Set<V> doInRedis(JedisCluster connection) {
				if(null == count || count == 0) return null;
				return toObjectSet(connection.spop(rawKey(key), count));
			}
		});
	}

	@Override
	public Long scard(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.scard(key);
			}
		});
	}

	@Override
	public Boolean sismember(final String key, final V member) {
		return executeCluster(new RedisCallback<JedisCluster, Boolean>() {
			@Override
			public Boolean doInRedis(JedisCluster connection) {
				if(null == member) return false;
				return connection.sismember(rawKey(key), raw(member));
			}
		});
	}
	
	@Override
	public Long smove(final String source, final String dest, final V member) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.smove(rawKey(source), rawKey(dest), raw(member));
			}
		});
	}

	@Override
	public Long zadd(final String key, final double score, final V member) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.zadd(rawKey(key), score, raw(member));
			}
		});
	}

	@Override
	public List<V> zrange(final String key, final Long start, final Long end) {
		return executeCluster(new RedisCallback<JedisCluster, List<V>>() {
			@Override
			public List<V> doInRedis(JedisCluster connection) {
				return toObjectList(connection.zrange(rawKey(key), start, end));
			}
		});

	}

	@Override
	public Long zrem(final String key, final V... member) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				if(null == member || member.length == 0) return 0l;
				return connection.zrem(rawKey(key), toByteArray(member));
			}
		});
	}

	@Override
	public Double zincrby(final String key, final double score, final V member) {
		return executeCluster(new RedisCallback<JedisCluster, Double>() {
			@Override
			public Double doInRedis(JedisCluster connection) {
				if(null == member) return null;
				return connection.zincrby(rawKey(key), score, raw(member));
			}
		});
	}

	@Override
	public Long zrank(final String key, final V member) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				if(null == member) return -1l;
				return connection.zrank(rawKey(key), raw(member));
			}
		});
	}

	@Override
	public Long zrevrank(final String key, final V member) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				if(null == member) return -1l;
				return connection.zrevrank(rawKey(key), raw(member));
			}
		});
	}

	@Override
	public List<V> zrevrange(final String key, final Long start, final Long end) {
		return executeCluster(new RedisCallback<JedisCluster, List<V>>() {
			@Override
			public List<V> doInRedis(JedisCluster connection) {
				return toObjectList(connection.zrevrange(rawKey(key), start, end));
			}
		});
	}

	@Override
	public Long zcard(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.zcard(key);
			}
		});
	}

	@Override
	public Double zscore(final String key, final V member) {
		return executeCluster(new RedisCallback<JedisCluster, Double>() {
			@Override
			public Double doInRedis(JedisCluster connection) {
				if(null == member) return null;
				return connection.zscore(rawKey(key), raw(member));
			}
		});
	}

	@Override
	public Long zcount(final String key, final double min, final double max) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.zcount(key, min, max);
			}
		});
	}

	@Override
	public List<V> zrangeByScore(final String key, final double min, final double max) {
		return executeCluster(new RedisCallback<JedisCluster, List<V>>() {
			@Override
			public List<V> doInRedis(JedisCluster connection) {
				return toObjectList(connection.zrangeByScore(rawKey(key), min, max));
			}
		});
	}

	@Override
	public List<V> zrevrangeByScore(final String key, final double max, final double min) {
		return executeCluster(new RedisCallback<JedisCluster, List<V>>() {
			@Override
			public List<V> doInRedis(JedisCluster connection) {
				return toObjectList(connection.zrevrangeByScore(rawKey(key), min, max));
			}
		});
	}

	@Override
	public Long zremrangeByRank(final String key, final Long start, final Long end) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.zremrangeByRank(key, start, end);
			}
		});
	}

	@Override
	public Long zremrangeByScore(final String key, final double start, final double end) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.zremrangeByScore(key, start, end);
			}
		});
	}


	@Override
	public Long decrBy(final String key, final Long integer) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.decrBy(key, integer);
			}
		});
	}

	@Override
	public Long decr(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.decr(key);
			}
		});
	}

	@Override
	public Long incrBy(final String key, final Long integer) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.incrBy(key, integer);
			}
		});
	}

	@Override
	public Long incr(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.incr(key);
			}
		});
	}

	@Override
	public void del(final String key) {
		executeCluster(new RedisCallback<JedisCluster, Void>() {
			@Override
			public Void doInRedis(JedisCluster connection) {
				connection.del(key);
				return null;
			}
		});

	}

	@Override
	public void expire(final String key, final int timeout, final TimeUnit unit) {
		executeCluster(new RedisCallback<JedisCluster, Void>() {
			@Override
			public Void doInRedis(JedisCluster connection) {
				connection.expire(rawKey(key), timeout * unit.getCount());
				return null;
			}
		});
	}

	@Override
	public Long ttl(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.ttl(key);
			}
		});
	}

	@Override
	public boolean exists(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, Boolean>() {
			@Override
			public Boolean doInRedis(JedisCluster connection) {
				return connection.exists(key);
			}
		});
	}

	@Override
	public Set<String> keys(final String pattern) {
		return executeCluster(new RedisCallback<JedisCluster, Set<String>>() {
			@Override
			public Set<String> doInRedis(JedisCluster connection) {
				Set<String> keys = new HashSet<>();  
				Map<String, JedisPool> clusterNodes = connection.getClusterNodes();  
				for(String k : clusterNodes.keySet()){  
					JedisPool jp = clusterNodes.get(k);  
					Jedis jedis = null;  
					try {  
						jedis = jp.getResource(); 
						keys.addAll(jedis.keys(pattern));  
					} finally{  
						if(jedis != null) jedis.close();
					}  
				}  
				return keys;  
			}
		});
	}

	@Override
	public Long publish(final String channel, final V message) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				return connection.publish(rawKey(channel), raw(message));
			}
		});
	}

	@Override
	public void subcribe(final Subcriber<V> subcriber,final String channel,final Pool<Jedis> pool) {
		executeCluster(new RedisCallback<JedisCluster, Void>() {
			@Override
			public Void doInRedis(final JedisCluster connection) {
				new Thread(new Runnable() {
					@Override
					public void run() {
						try {
							connection.subscribe(getListener(channel, subcriber), rawKey(channel));
						}catch (Exception e) {
							e.printStackTrace();
						} finally {
							if(connection != null) {
								try {
									connection.close();
								} catch (IOException e) {
									e.printStackTrace();
								}
							}
						}
					}
				}).start();

				return null;
			}
		});
	}

	@Override
	public void pipelineSet(final Map<String,V> hash) {
		executeCluster(new RedisCallback<JedisCluster, Void>() {
			@Override
			public Void doInRedis(JedisCluster connection) {
				if(null == hash || hash.isEmpty()) return null;
				JedisClusterPipeline jcp = new JedisClusterPipeline(connection);
				try {
					jcp.refreshCluster();
					for(String key : hash.keySet()) {
						jcp.set(rawKey(key),raw(hash.get(key)));
					}
					jcp.sync();
				}finally {
					jcp.close();
				}
				return null;
			}
		});
	}

	@Override
	public List<V> pipelineGet(final Collection<String> keys) {
		return executeCluster(new RedisCallback<JedisCluster, List<V>>() {
			@Override
			public List<V> doInRedis(JedisCluster connection) {
				List<V> data = new ArrayList<>();
				if(null != keys && !keys.isEmpty()) {
					JedisClusterPipeline jcp = null;
					try {
						jcp = new JedisClusterPipeline(connection);
						jcp.refreshCluster();
						for(String key : keys) {
							jcp.get(rawKey(key));
						}
						data = bytetoObjectList(jcp.syncAndReturnAll());
					}finally {
						if(jcp != null) jcp.close();
					}
				}

				return data;
			}
		});
	}

	@Override
	public void pipelineDel(final Collection<String> keys) {
		executeCluster(new RedisCallback<JedisCluster, Void>() {
			@Override
			public Void doInRedis(JedisCluster connection) {
				if(null == keys || keys.isEmpty())  return null;

				JedisClusterPipeline jcp = null;
				try {
					jcp = new JedisClusterPipeline(connection);
					jcp.refreshCluster();
					for(String key : keys) {
						jcp.del(key);
					}
					jcp.sync();
				}finally {
					if(jcp != null) jcp.close();
				}
				return null;
			}
		});
	}
	
	@Override
	public Long counter(final String key) {
		return executeCluster(new RedisCallback<JedisCluster, Long>() {
			@Override
			public Long doInRedis(JedisCluster connection) {
				byte[] bytes = connection.get(rawKey(key));
				if(bytes == null) return null;
				String count = new String(bytes);
				return count == null ? null : Long.parseLong(count);
			}
		});
	}

	@Override
	public void pipelineRpush(final Map<String, V> values) {
		executeCluster(new RedisCallback<JedisCluster, Void>() {
			@Override
			public Void doInRedis(JedisCluster connection) {
				if(null == values || values.isEmpty()) return null;
				JedisClusterPipeline jcp = new JedisClusterPipeline(connection);
				try {
					jcp.refreshCluster();
					for(String key : values.keySet()) {
						jcp.rpush(rawKey(key),raw(values.get(key)));
					}
					jcp.sync();
				}finally {
					jcp.close();
				}
				return null;
			}
		});
	}
	
	
}
